package cn.tedu.confirm;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

//异步发布确认模式
public class YiBuConfirm {
    public static void main(String[] args) throws Exception{
        //1.通过连接工厂配置rabbitmq的连接信息
        ConnectionFactory cf = new ConnectionFactory();
        cf.setHost("127.0.0.1");
        cf.setPort(5672); //客户端端口，5672是默认端口，可以省略
        cf.setUsername("guest");
        cf.setPassword("guest");
        //2.创建tcp和channel
        Connection nc = cf.newConnection();
        Channel cc = nc.createChannel();
        //3.使用随机队列名
        cc.queueDeclare("helloworld",false,false,false,null);
        //开启发布确认
        cc.confirmSelect();
        /*存储为确认的消息
           线程安全有序的一个hash表，适用于高并发的情况
           1.轻松将序号与消息进行关联
           2.支持并发访问
           3.只要给序列化，轻松批量删除
        */
        ConcurrentSkipListMap<Long,String> map = new ConcurrentSkipListMap<>();
        /*
          监听器的回调，确认收到的一个回调
          1.消息的序列号
          2.true 可以确认小于等于当前序列号的消息，false确认当前序列化的消息
            如果是true所有序列号不大于当前序列号的消息都会被确认
            如果是false，每次只能确认或者否认一条消息
         */
        ConfirmCallback ackCallback = new ConfirmCallback() {
            @Override
            public void handle(long l, boolean b) throws IOException {
                System.out.println("发送成功的消息编号："+l);
                if(b){
                    //返回的是小于或等于当前序列号的未确认消息
                    ConcurrentNavigableMap<Long,String> confirmed
                            = map.headMap(l,true);
                    //清除该部分未确认的消息
                    confirmed.clear();
                } else {
                    map.remove(l);
                }
            }
        };
        //发布失败的回调函数
        ConfirmCallback unAckCallback = new ConfirmCallback() {
            @Override
            public void handle(long l, boolean b) throws IOException {
                System.out.println(map.get(l)+"发布失败");
            }
        };
        long start = System.currentTimeMillis();
        for (int i = 1; i <=100 ; i++) {
            String msg = "消息:"+i;
            //cc.getNextPublishSeqNo()可以获取下一个消息的序列化
            //通过序列号与消息进行关系，全部都是未确认的消息
            map.put(cc.getNextPublishSeqNo(),msg);
            cc.basicPublish("","helloworld",null,msg.getBytes());
        }
        long end = System.currentTimeMillis();
        System.out.println("发送100消息数据使用的时间:"+(end-start));
    }
}
